package com.mengxin.secondkill.service.impl;

import com.mengxin.secondkill.constants.*;
import com.mengxin.secondkill.domain.Goods;
import com.mengxin.secondkill.domain.Order;
import com.mengxin.secondkill.domain.Stock;
import com.mengxin.secondkill.exception.ServiceException;
import com.mengxin.secondkill.mapper.GoodsMapper;
import com.mengxin.secondkill.mapper.StockMapper;
import com.mengxin.secondkill.message.redisqueue.BuyGoodsData;
import com.mengxin.secondkill.message.redisqueue.MsgData;
import com.mengxin.secondkill.message.req.AddGoodsReq;
import com.mengxin.secondkill.message.req.AddOrderReq;
import com.mengxin.secondkill.message.req.BuyGoodsReq;
import com.mengxin.secondkill.message.req.GoodsStockCacheReq;
import com.mengxin.secondkill.message.resp.AddGoodsResp;
import com.mengxin.secondkill.message.resp.AddOrderResp;
import com.mengxin.secondkill.message.resp.BuyGoodsResp;
import com.mengxin.secondkill.service.GoodsService;
import com.mengxin.secondkill.msgqueue.RedisMessageQueue;
import com.mengxin.secondkill.util.RedisUtil;
import com.mengxin.secondkill.util.SonwFlakeKeyGenerator;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @author mengxin
 * @date 2020/12/14 19:32
 */
@Service
@Slf4j
public class GoodsServiceImpl implements GoodsService {

    @Autowired
    GoodsMapper goodsMapper;

    @Autowired
    RedisTemplate<String,Object> redisTemplate;

    @Autowired
    StockMapper stockMapper;

    @Autowired
    RedisUtil redisUtil;

    @Autowired
    RedisMessageQueue redisMessageQueue;

    @Autowired
    SonwFlakeKeyGenerator sonwFlakeKeyGenerator;

    @Setter
    @Value("${redis.version}")
    private String version;

    private Long loclTimeOut = 2*1000L;

    @Override
    public AddGoodsResp addGoods(AddGoodsReq req) {
        AddGoodsResp resp = new AddGoodsResp();
        Goods goods = new Goods();
        BeanUtils.copyProperties(req, goods);
        goods.setCreateTime(LocalDateTime.now());
        //默认上线状态
        goods.setGoodsStatus(GoodsConstant.Status.ONLINE.getCode());
        //插入数据到数据库
        goodsMapper.insertSelective(goods);
        Long goodsId = goods.getGoodsId();
        //更新布隆过滤器
        addbloomFilter(goodsId);
        resp.setGoodsId(goodsId);
        return resp;
    }

    /**
     * @param goodsId
     */
    private void addbloomFilter(Long goodsId) {
        redisUtil.addByBloomFilter(BloomFilterConstant.BLOOM_FILTER, CacheConstant.GOODS_BLOOM_FILTER, goodsId);
    }

    @Override
    public BuyGoodsResp buyGoods(BuyGoodsReq req) throws InterruptedException {
        BuyGoodsResp resp = new BuyGoodsResp();
        Long goodsId = req.getGoodsId();
        //查询布隆过滤器
        if(!exitbloomFilter(goodsId)){
            throw new ServiceException(RespCode.COMMON_FAIL_CODE, "没有该商品");
        }
        //校验库存
        validStock(goodsId,req.getModelId(),req.getSum());
        //利用雪花算法生成订单流水号
        long orderNumber = sonwFlakeKeyGenerator.nextId();
        //发送消息
        sendMsg(req,orderNumber);
        resp.setOrderNumber(orderNumber);
        return resp;
    }

    /**
     * 校验库存
     * @param goodsId
     * @param modelId
     * @param sum
     * @throws InterruptedException
     */
    private void validStock(Long goodsId,Long modelId,Long sum) throws InterruptedException {
        String key = CacheConstant.GOODS_STOCK + goodsId;
        String lockKey = key+":"+modelId;
        boolean flag = false;
        Object stockCache = null;
        Long stockNum = null;
        do{
            //获取Redis分布式锁
            flag = redisTemplate.opsForValue().setIfAbsent(lockKey,version,loclTimeOut,TimeUnit.MILLISECONDS);
            if(flag){
                Long start = System.currentTimeMillis();
                stockCache = redisTemplate.opsForHash().get(key, String.valueOf(modelId));
                //获取商品库存缓存信息
                if (stockCache == null) {
                    Stock stock = stockMapper.selectByGoodsModelId(goodsId,modelId);
                    if(stock==null){
                        throw new ServiceException(RespCode.COMMON_FAIL_CODE, "该商品未配置库存");
                    }else{
                        stockNum = stock.getStock();
                        //防止此时有大量请求查询该商品库存导致查询数据库，造成缓存击穿
                        redisTemplate.opsForHash().put(CacheConstant.GOODS_STOCK + goodsId,String.valueOf(modelId),stockNum);
                    }
                }else{
                    stockNum = Long.valueOf(stockCache.toString());
                }
                if (stockNum == 0) {
                    if(version.equals(redisTemplate.opsForValue().get(lockKey))){
                        //删除分布式锁
                        redisTemplate.delete(lockKey);
                    }
                    throw new ServiceException(RespCode.COMMON_FAIL_CODE, "小主，已经卖光啦");
                }
                if (sum > stockNum) {
                    if(version.equals(redisTemplate.opsForValue().get(lockKey))){
                        //删除分布式锁
                        redisTemplate.delete(lockKey);
                    }
                    throw new ServiceException(RespCode.COMMON_FAIL_CODE, "库存不足");
                }else{
                    //预防提前过期，redis分布式锁被其他线程获取
                    if(version.equals(redisTemplate.opsForValue().get(lockKey))){
                        //redis扣减库存
                        redisTemplate.opsForHash().put(key,String.valueOf(modelId),stockNum-sum);
                        //删除分布式锁
                        redisTemplate.delete(lockKey);
                    }else{
                        flag = false;
                    }
                }
                log.info("商品库存校验与扣减花费时常："+(System.currentTimeMillis()-start));
            }else{
                Thread.sleep(500L);
            }
        }while(!flag);
    }

    /**
     * 向Redis（Stream）消息队列中发送消息
     * @param req
     */
    public void sendMsg(BuyGoodsReq req,long orderNumber){
        BuyGoodsData buyGoods = new BuyGoodsData();
        BeanUtils.copyProperties(req,buyGoods);
        buyGoods.setOrderNumber(orderNumber);
        MsgData msgData = new MsgData(buyGoods);
        log.info("Send message; key:{},data:{}", StreamConstans.GOODS_BUY_STREAM_KEY,msgData.toString());
        redisMessageQueue.sendMsgStream(StreamConstans.GOODS_BUY_STREAM_KEY,msgData);
    }


    /**
     * 判断布隆过滤器中是否存在商品
     * @param goodsId
     */
    private boolean exitbloomFilter(Long goodsId) {
        return redisUtil.includeByBloomFilter(BloomFilterConstant.BLOOM_FILTER, CacheConstant.GOODS_BLOOM_FILTER, goodsId);
    }

    @Override
    public void goodsStockCache(GoodsStockCacheReq req) {
        //获取库存信息
        List<Stock> stocks = stockMapper.getListByGoodsId(req.getIds());
        Map<Long, Map<String, Long>> stockMap = new HashMap<>(stocks.size());
        for (Stock stock : stocks) {
            Map<String, Long> map = stockMap.get(stock.getGoodsId());
            if (map == null) {
                map = new HashMap<>();
            }
            map.put(stock.getModelId().toString(), stock.getStock());
            stockMap.put(stock.getGoodsId(), map);
        }
        //将库存信息放入缓存
        for (Long goodsId : stockMap.keySet()) {
            String key = CacheConstant.GOODS_STOCK + goodsId;
            redisTemplate.opsForHash().putAll(key, stockMap.get(goodsId));
        }
    }
}
